package p.ithorns.framework.redis.stream;

import io.lettuce.core.resource.ClientResources;
import io.lettuce.core.resource.NettyCustomizer;
import io.netty.channel.Channel;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.hash.HashMapper;
import p.ithorns.framework.redis.stream.serializer.StreamKeySerializer;
import p.ithorns.framework.redis.stream.serializer.StreamValueSerializer;
import p.ithorns.framework.redis.stream.mapper.RedisObjectHashMapper;

/**
 * RedisStreamContext
 *
 * @author ithorns
 * @version 1.0.0
 * @date 2022/12/10 12:42
 */
@ComponentScan("p.ithorns.framework.redis.stream")
public class RedisStreamConfig {

    private final static Logger log = LoggerFactory.getLogger(RedisStreamConfig.class);

    private final NettyCustomizer customizer = new NettyCustomizer() {

        @Override
        public void afterChannelInitialized(Channel channel) {
            channel.pipeline()
                    .addLast(new IdleStateHandler(20, 0, 0))
                    .addLast(new ChannelDuplexHandler() {
                        @Override
                        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
                            // log.info("{} -- {}", ctx.channel().id(), evt);
                            if (evt instanceof IdleStateEvent ) {
                                IdleStateEvent idleEvent = (IdleStateEvent) evt;
                                // 读
                                if (idleEvent.state() == IdleState.READER_IDLE) {
                                    log.info("READER_IDLE");
                                    // ctx.disconnect();
                                }
                            } else {
                                super.userEventTriggered(ctx, evt);
                            }
                        }

                        @Override
                        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
                            log.info("ChannelInactive");
                            synchronized ("resetStreamLock") {
                                // List<Subscription> subscriptions = (List<Subscription>) SpringUtil
                                // .getBean("subscriptions", List.class);
//                                if (!CollectionUtil.isEmpty(subscriptions)) {
//                                    for (Subscription ss : subscriptions) {
//                                        if (!ss.isActive()) {
//                                            log.info("{} 失效了", ss);
//                                        }
//                                    }
//                                }
                            }
                            super.channelInactive(ctx);
                        }
                    });
        }
    };

    @Bean
    public RedisStreamTemplate<?> redisStreamTemplate(RedisConnectionFactory factory,
                                                      StreamKeySerializer keySerializer,
                                                      StreamValueSerializer<Object> valueSerializer,
                                                      HashMapper<Object, String, ?> hashMapper) {
        return new RedisStreamTemplate<>(factory, keySerializer, valueSerializer, hashMapper);
    }

    @Bean
    public HashMapper<Object, String, ?> hashMapper(StreamValueSerializer<Object> redisSerializer) {
        //return new ObjectHashMapper();
        return new RedisObjectHashMapper<>(redisSerializer);
    }

    @Bean(destroyMethod = "shutdown")
    public ClientResources clientResources() {
        return ClientResources.builder()
                .nettyCustomizer(customizer)
                .build();
    }

}